//package org.zfes.snowier.xyz.analysis.queue;
//
//import java.util.Map;
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.Executors;
//import java.util.concurrent.LinkedBlockingQueue;
//import java.util.concurrent.ScheduledExecutorService;
//import java.util.concurrent.ThreadFactory;
//import java.util.concurrent.ThreadPoolExecutor;
//import java.util.concurrent.TimeUnit;
//import java.util.concurrent.atomic.AtomicBoolean;
//
//import org.zfes.snowier.common.eventpub.BasicMsg;
//import org.zfes.snowier.common.eventpub.EventBusBucket;
//import org.zfes.snowier.core.util.ZAlert;
//import org.zfes.snowier.core.util.ZStrUtil;
//
//import com.google.common.collect.Maps;
//import com.google.common.util.concurrent.ThreadFactoryBuilder;
//
//public class MemoryQueueBucket2 implements QueueBucket{
//
//	private static Map<String,SnowierQueue> queueBucket = Maps.newConcurrentMap();
//	
//	private static ExecutorService postExecutorService;
//	
//	
//	private static final int MAX_QUEUE_LENGTH = 100;
//	
//	private volatile  AtomicBoolean runScheduled = new AtomicBoolean(false);
//	 
//	/**
//	 * 0.避免while-true检测，让队列有适当消息堆积能力
//	 * 
//	 * 1.固定时间频率
//	 * 
//	 * 2.队列超过长度
//	 */
//	private static ExecutorService queueMonitorExecutorService;
//	
//	
//	private static ExecutorService offerExecutorService;
//	
//	private static final int maxOfferThreads=6;
//	
//	private static final int maxOfferThreadQueueSize=1024*1024;
//	
//	
//	private static AtomicBoolean sheduelPoolStarted = new AtomicBoolean(false);
//	
//	
//	private  SnowierQueue getQueue(String topic) {
//		startPostSheduel();
//		
//		if(ZStrUtil.hasNoText(topic)) {
//			ZAlert.appRuntimeError("topic can not be null ");
//		}
//		
//		SnowierQueue queue=queueBucket.get(topic);
//		if(queue==null) {
//				queue=new SnowierQueue(topic);
//				queueBucket.put(topic, queue);
//		}
//		return queue;
//	}
//	
//	
//	@Override
//	public  void offer(String topic,Object msg) {
//		SnowierQueue queue = getQueue(topic);
//		getOfferPool().submit(()->queue.offer(msg));
//	}
//	
//	@Override
//	public  void clear(String topic) {
//		 getQueue(topic).clear();
//	}
//	
//	private  void startPostSheduel(){
//		if(sheduelPoolStarted.get()) {
//			return;
//		}
//		if(postExecutorService==null) {
//			postExecutorService=Executors.newSingleThreadExecutor();
//		}
//		postExecutorService.submit(()->{
//			while(true) {
//				try {
//					queueBucket.keySet().forEach(key->{
//						SnowierQueue queue=getQueue(key);
//						
//						doPoll( queue, key);
//						
//						
//					});
//				}catch (Exception e) {
//					e.printStackTrace();
//				}
//			}
//		});
//		
//		
//		sheduelPoolStarted.set(true);
//	
//	}
//	
//	private void doPoll(SnowierQueue queue,String key) {
//		Object data=queue.poll();
//		if(data!=null) {
//			EventBusBucket.post(key, new BasicMsg(data).setTopic(queue.getTopic()));
//		}
//	}
//	
//	private static ExecutorService  getOfferPool() {
//		if(offerExecutorService != null) {
//			return offerExecutorService;
//		}
//		ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("-queueBucket-OfferPool-pool-%d").build();
//		offerExecutorService = new ThreadPoolExecutor(1, maxOfferThreads,0L, 
//						TimeUnit.MILLISECONDS,
//						new LinkedBlockingQueue<Runnable>(maxOfferThreadQueueSize), 
//						namedThreadFactory, 
//						new ThreadPoolExecutor.AbortPolicy());
//		return offerExecutorService;				
//		
//	}
//	
//}
